package com.mjf.spark.day05

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 行动算子
 *    -reduce
 *        聚集RDD中所有元素，先聚合分区内数据，再聚合分区间数据
 *    -collect
 *        在driver中，以数组Array的形式返回数据集的所有元素
 *    -foreach
 *        遍历RDD中每一个元素，直接在executor端完成
 *    -count
 *        获取RDD中元素的个数
 *    -first
 *        返回RDD中的第一个元素
 *    -take
 *        返回RDD中前n个元素组成的数组
 *    -takeOrdered
 *        获取RDD排序后前n个元素组成的数组
 *    -saveAsTextFile
 *        保存为文本文件
 *    -saveAsObjectFile
 *        保存序列化文件
 *    -saveAsSequenceFile
 *        保存为saveAsSequenceFile  注意：只支持pairRDD
 */
object Spark01_Action {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark01_Action")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(List(1, 6, 5, 2, 3, 4))

    // reduce
//    val resRDD: Int = rdd.reduce(_ + _)
//    println(resRDD)

    // collect
//    val resRDD: Array[Int] = rdd.collect()
//    resRDD.foreach(println)

    // foreach
//    rdd.foreach(println)

    // count
//    val res: Long = rdd.count()
//    println(res)

    // first
//    val first: Int = rdd.first()
//    println(first)

    // take
//    val res: Array[Int] = rdd.take(3)
//    println(res.mkString(","))

//    val res: Array[Int] = rdd.takeOrdered(3)
//    println(res.mkString(","))


    // save相关的算子
    // 保存为文本文件
    rdd.saveAsTextFile("D:\\code\\spark\\output")
    // 保存序列化文件
//    rdd.saveAsObjectFile("D:\\code\\spark\\output")
    // 保存为saveAsSequenceFile  注意：只支持pairRDD
//    rdd.map((_, 1)).saveAsSequenceFile("D:\\code\\spark\\output")

    // 关闭连接
    sc. stop()

  }
}
